package fregat.connector.worker;

import fregat.base.database.DatabaseConnector;
import fregat.base.util.StopWatch;
import fregat.connector.ConnectorService;

import java.sql.*;

/**
 * Worker for database alert triggered processes.
 *
 * @author Alexander Mann <br><b><i>ECO_L O G</i> &nbsp Logistiksysteme GmbH</b>
 * @version 1.0
 */
public class AlertWorker extends DatabaseWorker {
  protected static final String SHUTDOWN = "SHUTDOWN"; // alert message for shutdown
  protected static final String START = "START"; // alert message for start

  protected String retrieveKey = "retrieve"; // key for the retrieve fetch
  protected String getKey = "get"; // key for the get fetch
  protected String dataKey = "data"; // key for the data fetch
  protected String updateKey = "update"; // key for the update of transfer state

  protected String retrieveSQL; // SQL string for retrieving IDs of data to be processed from database
  protected String dataSQL; // SQL string for retrieving IDs of data to send from database
  protected String getSQL; // SQL string for getting send data from database
  protected String updateSQL; // SQL string for updating transfer state

  protected long dataID; // ID of currently transferred data
  protected boolean retrieveConfigured; // retrieve statement is configured
  private boolean getFunction; // function is used for get operation

  private String alertName; // name of database alert
  private int alertTimeout; // time to wait for alert

  /**
   * Configures the database alert.
   *
   * @param name Configuration name of worker
   */
  public AlertWorker(String name) {
    super(name);

    // configure alert -->
    String configPath = "worker.config." + getName() + ".db.alert";
    alertName = configuration.getConfigurationString(configPath + ".name");
    alertTimeout = configuration.getConfigurationInteger(configPath + ".timeout");

    if (alertTimeout < 0) {
      alertTimeout = 30;
    }
    // <--

    // configure table and types -->
    configPath = "worker.config." + getName() + ".db";
    String tableName = configuration.getConfigurationString(configPath + ".table");
    if (tableName == null) {
      tableName = "SSTDAT";
    }
    String dataTypes = configuration.getConfigurationString(configPath + ".types");
    // <--

    // configure select to fetch IDs of data to be sent -->
    dataSQL = getSQL(configPath + ".data");
    if (dataSQL == null && dataTypes != null) {
      dataSQL = "select SSTDAT_ID from " + tableName +
          " where SSTTYP in (" + dataTypes + ") and SSTSTAT <= 1 order by SSTDAT_ID";
    } else {
      ConnectorService.warn(ConnectorService.resource.getString("AlertWorker_constructor_1").replaceFirst("%1", name));
    }
    // <--

    // configure update of transfer result -->
    if (configuration.getConfigurationBoolean(configPath + ".update") && dataTypes != null) {
      // use default statement
      updateSQL = "update " + tableName + " set SSTSTAT = ? where SSTDAT_ID = ?";
      updateKey += " -> " + updateSQL;

    } else {
      // statement or procedure
      updateSQL = getSQL(configPath + ".update");
      if (updateSQL != null) {
        // procedure is configured
        updateKey += " -> " + updateSQL + "(?,?)";
        updateSQL = "begin " + updateSQL + "(?,?); end;";
      }
    }
    // <--

    // configure select to fetch IDs of data to be processed -->
    retrieveSQL = getSQL(configPath + ".retrieve");
    if (retrieveSQL == null) {
      retrieveSQL = dataSQL;
    } else {
      retrieveConfigured = true;
    }

    if (retrieveSQL != null) {
      retrieveKey += " -> " + retrieveSQL;
      retrieveSQL = setSelectID(retrieveSQL, "ConnectorService,AlertWorker,retrieve");
    } else {
      ConnectorService.error(ConnectorService.resource.getString("AlertWorker_constructor_2"));
      ConnectorService.fatalError();
    }
    // <--

    // set select ID now as data statement may also be used for retrieving
    if (dataSQL != null) {
      dataKey += " -> " + dataSQL;
      dataSQL = setSelectID(dataSQL, "ConnectorService,AlertWorker,data");
    }

    // configure select to get single data item to send from database -->
    getSQL = getSQL(configPath + ".get");
    if (getSQL == null) {
      // use default statement
      getSQL = "select NUTZDATEN from " + tableName + " where SSTDAT_ID = ?";
    } else {
      // statement or function
      if (getSQL.toUpperCase().startsWith("SELECT")) {
        // statement is configured
        getKey += " -> " + getSQL;
        getSQL = setSelectID(getSQL, "ConnectorService,AlertWorker,get");
      } else {
        // function is configured
        getKey += " -> " + getSQL + "(?)";
        getSQL = "begin ? := " + getSQL + "(?); end;";
        getFunction = true;
      }
    }
    // <--
  }

  /**
   * The configured alert was fired in database with the specified message.
   *
   * @param message Alert message
   */
  protected boolean alertReceived(String message) {
    boolean processing = false;

    try {
      // get IDs of entries that should be processed
      ResultSet processIDs = getProcessIDs();

      while (!shutdown && processIDs.next() && connected) {
        // process data
        processing = true;
        long processID = processIDs.getLong(1);
        boolean verified = false;

        while (!verified && !shutdown && connected) {
          ConnectorService.debug(getName() + ", " + ConnectorService.resource.getString("AlertWorker_alertReceived_1") +
              " #" + processID);
          if (processSQL != null) {
            // process data
            processData(processID, message);

            // commit changes
            databaseConnector.commit();

            // transfer data
            verified = transferData();

          } else {
            // no processing procedure configured --> retrieved data is transfer data
            dataID = processID;
            verified = transferData(getData(dataID));
          }
        }
      }

      if (!isStandalone()) {
        // inform channel that data was completely processed
        ConnectorService.dataComplete(channelName);
      }

    } catch (SQLException ex) {
      databaseConnector.rollback();
      ConnectorService.error(getName() + ", AlertWorker.alertReceived", ex);
    }

    return processing;
  }

  /**
   * Collects the IDs of entries to transfer and transfers each data entry using
   * {@link boolean transferData(java.lang.String)}.
   *
   * @return True, if all entries have been successfully transferred
   */
  protected boolean transferData() {
    if (dataSQL == null) {
      // no data statement configured, no data is transferred
      return true;

    } else {
      try {
        ResultSet dataIDs = getDataIDs();
        boolean success = true;
        boolean processing = false;

        while (dataIDs.next() && !shutdown && success) {
          processing = true;

          dataID = dataIDs.getInt(1);

          ConnectorService.debug(getName() + ", " +
              ConnectorService.resource.getString("AlertWorker_transferData_1") +
              " #" + dataID);

          success = transferData(getData(dataID));

          if (updateSQL != null) {
            CallableStatement statement = databaseConnector.getStatement(updateKey);
            if (statement == null) {
              statement = databaseConnector.prepareStatement(updateKey, updateSQL);
            }
            int transferState = success ? 2 : 3;
            statement.setInt(1, transferState);
            statement.setLong(2, dataID);
            executeStatement(DatabaseConnector.logStatement(updateKey, new String[]{Integer.toString(transferState), Long.toString(dataID)}), updateKey);
            databaseConnector.commit();
          }
        }

        return success && processing;

      } catch (SQLException ex) {
        ConnectorService.error("AlertWorker.transferData", ex);
        databaseConnector.rollback();
      }
      return false;
    }
  }

  /**
   * Transfers the given data. The default implementation is to insert the
   * data using the configured 'insert' function.
   *
   * @param data Data to transfer
   * @return True, if success
   */
  protected boolean transferData(String data) {
    try {
      long value = insertData(channelName, data);

      if (value > 0) {
        databaseConnector.commit();
        return true;
      } else {
        databaseConnector.rollback();
      }

    } catch (SQLException ex) {
      ConnectorService.error("AlertWorker.transferData(String)", ex);
      databaseConnector.rollback();
    }
    return false;
  }

  /**
   * Connects to the database, calls <code>alertReceived</code> once to send all
   * data waiting in database and waits for database alert.
   */
  public void start() {
    super.start();
    databaseConnector.connectToDatabase();
    databaseConnector.setName(ConnectorService.VERSION);
    databaseConnector.setAction("Alert");

    if (databaseConnector.isConnected() && alertName != null) {
      new AlertThread().start();

      // wait some time to ensure alert is registered
      try {
        Thread.sleep(1000);
      } catch (InterruptedException ex) {
      }
    }
  }

  /**
   * Stops waiting for database alert and disconnects from database.
   */
  public void shutdown() {
    super.shutdown();
    if(alertName != null) {
      databaseConnector.fireAlert(alertName, SHUTDOWN);
    }

    if (!databaseConnector.isConnected()) {
      databaseConnector.shutdown();
    }
  }

  /**
   * Fetches data that has to be processed by executing the configured
   * process statement.g
   *
   * @return Data to be processed
   * @throws SQLException
   */
  protected ResultSet getProcessIDs() throws SQLException {
    return fetchResult(retrieveKey, retrieveSQL);
  }

  /**
   * Fetchtes data that has to be sent by executing the configured data
   * statement.
   *
   * @return Data to be sent
   * @throws SQLException
   */
  protected ResultSet getDataIDs() throws SQLException {
    return fetchResult(dataKey, dataSQL);
  }

  /**
   * Fetches data to send by executing the configured get statement.
   *
   * @param id ID of data to send
   * @return Data to send
   * @throws SQLException
   */
  protected String getData(long id) throws SQLException {
    CallableStatement statement = databaseConnector.getStatement(getKey);
    if (statement == null) {
      statement = databaseConnector.prepareStatement(getKey, getSQL);
      if (getFunction) {
        statement.registerOutParameter(1, Types.VARCHAR);
      }
    }

    if (getFunction) {
      // process get operation as function
      statement.setLong(2, id);
      executeStatement(DatabaseConnector.logStatement(getKey, new String[]{Long.toString(id)}), getKey);
      databaseConnector.commit();
      return statement.getString(1);

    } else {
      // process get operation as statement
      statement.setLong(1, id);
      ResultSet result = executeQuery(DatabaseConnector.logStatement(getKey, new String[]{Long.toString(id)}), getKey);
      if (result.next()) {
        return result.getString(1);
      }
    }


    return null;
  }

  public void connected(boolean b) {
    super.connected(b);
    if (b) {
      if(alertName != null) {
        databaseConnector.fireAlert(alertName, START);
      } else {
        alertReceived(START);
      }
    }
  }

  /**
   * Looks up the SQL command in the configuration file. If the command is
   * specified with a numeric value the SQL command is looked up in the SQL
   * table using the numeric value as SQL_ID.
   *
   * @param configPath Path to configuration key
   * @return SQL command or null if none found
   */
  protected String getSQL(String configPath) {
    String sql = configuration.getConfigurationString(configPath);

    if (sql != null && !sql.toUpperCase().trim().startsWith("SELECT")) {
      // SQL ID configured
      int sqlId = configuration.getConfigurationInteger(configPath,false);

      if (sqlId > 0) {
        // try to find select in SQL table
        Connection connection = databaseConnector.createConnection(false, true);

        if (connection != null) {
          try {
            CallableStatement statement = connection.prepareCall("select STATEMENT from SQL where SQL_ID = ?");
            statement.setInt(1, sqlId);

            ResultSet result = statement.executeQuery();

            if (result.next()) {
              sql = result.getString("STATEMENT");

              System.out.println(sql);
            } else {
              sql = null;
            }

            connection.close();

          } catch (SQLException ex) {
            ConnectorService.error("AlertWorker.getSQL", ex);
            sql = null;
          }
        } else {
          ConnectorService.warn(ConnectorService.resource.getString("AlertWorker_getSQL_1").replaceFirst("%1", configPath));
          sql = null;
        }
      }
    }

    return sql;
  }

  /**
   * Thread that waits for database alert and calls <code>alertReceived</code>
   * when alert is received or timeout expired.
   */
  private class AlertThread extends Thread {

    public void processAlert(String message) {
      ConnectorService.note(AlertWorker.this.getName() + ", '" + message + "'");
      StopWatch.resetTime("alert");
      if(alertReceived(message)) {
        ConnectorService.debug(AlertWorker.this.getName() + ", " +
            ConnectorService.resource.getString("AlertWorker.AlertThread_run_1") +
            " " + StopWatch.takeTime("alert") + "[ms]");
      }
    }

    public void run() {      
      while (!shutdown) {
        if (isStandalone() && !connected) {
          connected = true;
          alertReceived(START);
        }

        // wait for alert
        String message = databaseConnector.waitForAlert(alertName, alertTimeout);

        if (!shutdown && message != null && !message.equals(SHUTDOWN)) {
          if (connected) {
            // process the alert's message
            processAlert(message);

          } else {
            // channel is not connected
            ConnectorService.note(ConnectorService.resource
                .getString("AlertWorker.AlertThread_run_2")
                .replaceFirst("%1", channelName));
          }
        }

        if (message == null) {
          // alert couldn't register - wait before retry
          try {
            Thread.sleep(2000);
          } catch (InterruptedException ex) {
            ex.printStackTrace();
          }

        }
      }
      databaseConnector.shutdown();
      ConnectorService.note(AlertWorker.this.getName() + ", AlertThread finished");
    }
  }
}